package streaming.api.tableapi;

import streaming.api.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import utils.PropertiesReader;

/**
 * 从文件读取数据流
 * DataStream转成Table （fromDataStream）
 * Table转成DataStream 转换有两种转换模式：追加（Appende）模式和撤回（Retract）模式
 */
public class TableTest1 {

    private static String filePath = PropertiesReader.get("default.file.from.path");

    public static void main(String[] args) throws Exception {
        // 1. 创建表环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000L);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 2. 读取数据
        DataStream<String> inputStream = env.readTextFile(filePath);
        // 3. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
//        dataStream.print("dataStream:");
        // 4. 基于流创建表 流DataStream 转换为 表Table
        Table dataTable = tableEnv.fromDataStream(dataStream);
        // 流 + Expression
        //Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature");
        // 输出表结构
        dataTable.printSchema();

        // 5. 调用table API    条件查询-根据ID
        Table resultTable = dataTable.select("id, temperature").where("id = 'sensor_a'");
        // Table 转换为 DataStream 输出展示
        //tableEnv.toAppendStream(resultTable, Row.class).print("result");

        // 6. 执行自定义SQL    条件查询-根据ID
        // 创建临时视图(Temporary View)
        tableEnv.createTemporaryView("sensor", dataTable);
        String sql = "select id, temperature, pt.proctime from sensor where id = 'sensor_a'";
        // tableEnv.sqlQuery(sql);
        // Table 转换为 DataStream 输出展示
        tableEnv.toAppendStream(tableEnv.sqlQuery(sql), Row.class).print("sql");

        // 7. 执行
        env.execute();
    }

}
